package com.learn.flink.learn.transformation;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SelfPartitionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<Long> dataStreamSource = env.addSource(new MySource());
        SingleOutputStreamOperator<Tuple1<Long>> map = dataStreamSource.map(new MapFunction<Long, Tuple1<Long>>() {
            @Override
            public Tuple1<Long> map(Long value) throws Exception {
                System.out.println("线程ID" + Thread.currentThread().getId() + "; value:" + value);
                return Tuple1.of(value);
            }
        });
        DataStream<Tuple1<Long>> dataStream = map.partitionCustom(new MyPartitioner(), 0);
        SingleOutputStreamOperator<Long> result = dataStream.map(new MapFunction<Tuple1<Long>, Long>() {
            @Override
            public Long map(Tuple1<Long> value) throws Exception {
                System.out.println("线程ID" + Thread.currentThread().getId() + "; value:" + value);
                return value.getField(0);
            }
        });
        result.print().setParallelism(1);
        env.execute("selfPartitionDemo");
    }
}
